
#include "elog.h"
#include "ipc_event.h"
#include "ipc_event_types.h"
#include "ipc_event_util.h"
#include <iostream>
#include <stdio.h>

#include <stdlib.h>

using namespace ipc_event;
using namespace std;

#define TAG "IPC_APP_interprocess"

// const value definition
static const int NULL_SOCKET = -1;
static const int NULL_EVT_CHANNEL_ID = -1;
static const int MESSAGE_HDR_LENGTH = 12;
static const int MESSAGE_BD_MAX_LENGTH =1024*4;/// 102400;//100kb
static const char * DEFAULT_IPC_EVT_IP = "127.0.0.1";

// the status of the working thread
static bool isRunning = false;

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
IPCPeer::IPCPeer():socket_m(NULL_SOCKET) , status_m(IPC_PEER_UNKNOWN)
{}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
IPCPeer::~IPCPeer()
{
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
void IPCPeer::setSocket(int fd)
{
  socket_m = fd;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
int IPCPeer::getSocket() const
{
  return socket_m;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
void IPCPeer::setPeerStatus(PeerStatus_t status)
{
  status_m = status;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
PeerStatus_t IPCPeer::getPeerStatus() const
{
  return status_m;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
IPCEventChannel::IPCEventChannel():eventChannelId_m(NULL_EVT_CHANNEL_ID)
{
  (void)subscribers_m.clear();
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
IPCEventChannel::~IPCEventChannel()
{
  eventChannelId_m = NULL_EVT_CHANNEL_ID;
  (void)subscribers_m.clear();
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
IPCSubscribers_t & IPCEventChannel::getSubscribers()
{
  return subscribers_m;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
void IPCEventChannel::setEventChannelId(int evtChannelId)
{
  eventChannelId_m = evtChannelId;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
int IPCEventChannel::getEventChannelId() const
{
  return eventChannelId_m;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCEventChannel::addSubscriber(int subscriber)
{
  for (IPCSubscribers_t::iterator it = subscribers_m.begin();
    it != subscribers_m.end();it ++)
  {
    if (*it == subscriber)
    {
      return false;
    }
  }
  
  (void)subscribers_m.push_back(subscriber);
  
  return true;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCEventChannel::delSubscriber(int subscriber)
{
  for (IPCSubscribers_t::iterator it = subscribers_m.begin();
    it != subscribers_m.end();it ++)
  {
    if (*it == subscriber)
    {
      (void)subscribers_m.erase(it);
      
      return true;
    }
  }
  
  return false;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
IPCPeerManager * IPCPeerManager::instance_m = 0;

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
IPCPeerManager  * IPCPeerManager::instance()
{
  if (instance_m != 0)
  {
    return instance_m;
  }
  
  instance_m = new IPCPeerManager;
  
  return instance_m;
}


IPCPeerManager::IPCPeerManager()
:listenSock_m(-1)
,port_m(0)
{
}


/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::initialize(unsigned short port)
{
  port_m = port;
  
  (void)channels_m.clear();
  (void)peers_m.clear();

  printf("IPCPeerManager start to initialize!\n");
  
  listenSock_m = socket(AF_INET , SOCK_STREAM , 0);//创建服务器socket
  if (listenSock_m == -1)
  {
    LOGE(TAG, "socket create failed(%s)!!\n", strerror(errno));
    return false;
  }
  
  struct sockaddr_in addr;
  (void)memset(&addr , 0x00 , sizeof(addr));
  /* sin_addr存放socket的ip地址，sin_port存放socket的端口号 */
  addr.sin_family = AF_INET; 
  addr.sin_port = htons(port_m);
  addr.sin_addr.s_addr = inet_addr(DEFAULT_IPC_EVT_IP);

  printf("IPCPeerManager start to bind!\n");
  
  if (bind(listenSock_m , (struct sockaddr *)&addr , sizeof(addr)) == -1)//使用bind将其绑定
  {
    LOGE(TAG, "socket bind failed(%s)!!\n", strerror(errno));
    (void)close(listenSock_m);
    listenSock_m = -1;
    
    return false;
  }

  printf("IPCPeerManager start to listen!\n");
  
  if (listen(listenSock_m , 5) == -1)//需要开始监听了
  {
    LOGE(TAG, "socket listen failed(%s)!!\n", strerror(errno));
    (void)close(listenSock_m);
    listenSock_m = -1;
    
    return false;
  }
  
  return true;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
void IPCPeerManager::updateFDs(fd_set * fds)
{
  FD_ZERO(fds);//每次循环都要清空集合，否则不能检测描述符变化
  FD_SET(listenSock_m , fds);//添加描述符
  
  for (IPCPeers_t::iterator it = peers_m.begin();it != peers_m.end();it ++)
  {
    IPCPeer peer = *it;		
    FD_SET(peer.getSocket() , fds);//添加描述符
  }
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
static void * __runEventLoop(void * args)
{
  isRunning = true;
  
  while (isRunning)
  {
    fd_set fds_m;
    
    IPCPeerManager::instance()->updateFDs(&fds_m);//添加描述符信息
    
    struct timeval tv;
    tv.tv_sec =4;/// 2;
    tv.tv_usec = 0;
    /* 拿到最大的文件描述符 */
    int ret = select(IPCPeerManager::instance()->getMaxSubscriber() + 1 , &fds_m , 0 , 0 , &tv);
    if (ret == -1)
    {
      if (errno == EINTR)
      {
        continue;	
      }
      sleep(1);
      //break;
    }
    else if (ret == 0)
    {
      continue;
    }
    else
    {	/*检查fdset联系的文件句柄fd是否可读写，>0表示可读写*/
      if (FD_ISSET(IPCPeerManager::instance()->getHandle() , &fds_m))
      {
        if (!IPCPeerManager::instance()->addPeer())//创建与客户端连接，并把fd句柄放到容器中
        {
          break;
        }
      }
      else
      {
        IPCSubscribers_t subscribers = IPCPeerManager::instance()->getSubscribers();
        for (IPCSubscribers_t::iterator it = subscribers.begin();it != subscribers.end();it ++)
        {
          if (FD_ISSET(*it, &fds_m))
          {
            (void)IPCPeerManager::instance()->handleRequest(*it);
          }
        }
      }
    }
  }
  
  isRunning = false;
  
  (void)IPCPeerManager::instance()->finalize();
  
  return 0;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
void IPCPeerManager::runEventLoop()
{
  (void)__runEventLoop(0);
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::finalize()
{
  if (listenSock_m != -1)
  {
    (void)close(listenSock_m);
    listenSock_m = -1;
  }
  
  (void)channels_m.clear();
  (void)peers_m.clear();
  
  return true;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
int IPCPeerManager::getHandle() const
{
  return listenSock_m;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
int IPCPeerManager::getMaxSubscriber()
{
  int max = -1;
  if (listenSock_m > max)
  {
    max = listenSock_m;
  }
  
  for (IPCPeers_t::iterator it = peers_m.begin();it != peers_m.end();it++)
  {
    IPCPeer peer = *it;
    if (max < peer.getSocket())
    {
      max = peer.getSocket();
    }
  }
  
  return max;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
IPCSubscribers_t IPCPeerManager::getSubscribers()
{
  IPCSubscribers_t subscribers;
  
  for (IPCPeers_t::iterator it = peers_m.begin();it != peers_m.end();it++)
  {
    IPCPeer peer = *it;
    
    (void)subscribers.push_back(peer.getSocket());
  }
  
  return subscribers;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::addPeer()//创建与客户端连接，并把fd句柄 放到容器中
{
  struct sockaddr_in addr;
  (void)memset(&addr , 0x00 , sizeof(addr));
  socklen_t len = sizeof(addr);
  
  /* 如果监听到客户端连接，则调用accept接收这个连接并同时新建一个socket来和客户进行通信 */
  int fd = accept(listenSock_m , (struct sockaddr *)&addr , &len);
  if (fd == -1)
  {
    LOGE(TAG, "socket accept failed(%s)!!\n", strerror(errno));
    return false;
  }
  /* 添加进程信息 */
  IPCPeer peer;
  peer.setPeerStatus(IPC_PEER_ACTIVE);
  peer.setSocket(fd);
  
  (void)peers_m.push_back(peer);
  
  return true;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::handleRequest(int fd)//读取请求、应答消息内容
{
  char msgHdr[MESSAGE_HDR_LENGTH] = {0};
  int bytes = 0,ErrorV=0;
  do
  {
    int ret = recv(fd , msgHdr + bytes, MESSAGE_HDR_LENGTH - bytes, 0);
    if (ret <= 0)
    {
    	ErrorV=errno;
      if (ErrorV != EINTR&&ErrorV !=EAGAIN )
      {
        LOGE(TAG, "socket recv failed(%s)!!\n", strerror(errno));
        //ITE_LOG_PRINT(LOG_ENABLE, "socket receive error, delete peer from list,errno:" << errno);
        (void)delPeer(fd);
        (void)updateChannel(fd);
        (void)close(fd);
        printf("----handleRequest error2!-%d-%d-\n",errno,isRunning);
	//apollo:20190820-tmp add to reboot
#if 0//D_DEBUG_MODE==0
	if(isRunning)///&&(errno==0||errno == ENOENT))
	{
		system("sync");
		sleep(1);
		system("reboot");
	}
#endif
        return false;
      }
      else 
      {
        continue;
      }
    }
    bytes += ret;
  } while (bytes < MESSAGE_HDR_LENGTH);
    
  int length = -1;
  int cmd = -1;
  int type = -1;
  
  int offset = 0;
  offset += IPCMsgUtils::parse(msgHdr , offset , length);
  offset += IPCMsgUtils::parse(msgHdr , offset , cmd);
  offset += IPCMsgUtils::parse(msgHdr , offset , type);
  
  if (length <= 0 || length > MESSAGE_BD_MAX_LENGTH || 
    cmd <= IPC_EVENT_TYPE_UNKNOWN || cmd >= IPC_EVENT_TYPE_MAX)
  {
    #if 0
    ITE_LOG_PRINT(LOG_ENABLE, "recv invalid data,delete peer from list,length:" << length
			 << ",cmd:" << cmd
	 	 	 << ",type:" << type
	 	 	 );
	#endif
    (void)delPeer(fd);
    (void)updateChannel(fd);
    (void)close(fd);
	printf("----handleRequest error1!----\n");
    return false;
  }
  
  char msgBody[MESSAGE_BD_MAX_LENGTH] = {0};
  if ((length - MESSAGE_HDR_LENGTH) > 0)
  {
    bytes = 0;
    do
    {
      int ret = recv(fd , msgBody + bytes , length - MESSAGE_HDR_LENGTH - bytes , 0);
      if (ret <= 0)
      {
      	ErrorV=errno;
        if (ErrorV != EINTR&&ErrorV !=EAGAIN )
        {
          LOGE(TAG, "socket recv failed(%s)!!\n", strerror(errno));
          //ITE_LOG_PRINT(LOG_ENABLE, "socket receive error, delete peer from list,errno:" << errno);
          (void)delPeer(fd);
          (void)updateChannel(fd);
          (void)close(fd);
          printf("----handleRequest error0!----\n");
  
          return false;
        }
        else
        {
          continue;
        }
      }
    
      bytes += ret;
    } while (bytes < (length - MESSAGE_HDR_LENGTH));
  }
    
  switch (cmd)
  {
    case IPC_EVENT_TYPE_PUBLISH:
      return handlePubRequest(fd ,type , msgBody , length - MESSAGE_HDR_LENGTH);
    case IPC_EVENT_TYPE_SUBSCRIBE:
      return handleSubRequest(fd , type);
    case IPC_EVENT_TYPE_UNSUBSCRIBE:
      return handleUnsubRequest(fd , type);
    default:
	  return false;
  }
  
  return false;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::delPeer(int fd)
{
  for (IPCPeers_t::iterator it = peers_m.begin(); it != peers_m.end(); it ++)
  {
    IPCPeer peer = *it;
    if (peer.getSocket() == fd)
    {
      peers_m.erase(it);
      return true;
    }
  }
  
  return false;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::addChannel(int evtType , int fd)
{
  for (IPCEventChannels_t::iterator it = channels_m.begin(); it != channels_m.end();it ++)
  {
    IPCEventChannel & channel = *it;
    if (channel.getEventChannelId() == evtType)
    {
      // the channel is exsited
      channel.addSubscriber(fd);
      
      return true;
    }
  }
  
  // add a new one into the channel set.
  IPCEventChannel chann;
  chann.setEventChannelId(evtType);
  chann.addSubscriber(fd);
  channels_m.push_back(chann);
  
  return true;	
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::updateChannelByType(int channelId , int fd)
{
  for (IPCEventChannels_t::iterator it = channels_m.begin(); it != channels_m.end();it ++)
  {
    IPCEventChannel & channel = *it;
    if (channel.getEventChannelId() == channelId)
    {
      IPCSubscribers_t & subscribers = channel.getSubscribers();
      for (IPCSubscribers_t::iterator pit = subscribers.begin(); pit != subscribers.end();pit ++)
      {
        if (fd == *pit)
        {
          subscribers.erase(pit);
          
          return true;
        }
      }
    }
  }
  
  return false;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::updateChannel(int fd)
{
  for (IPCEventChannels_t::iterator it = channels_m.begin(); it != channels_m.end();it ++)
  {
    IPCEventChannel & channel = *it;
    IPCSubscribers_t & subscribers = channel.getSubscribers();
    for (IPCSubscribers_t::iterator pit = subscribers.begin(); pit != subscribers.end();pit ++)
    {
      if (fd == *pit)
      {
        subscribers.erase(pit);
        
        break;
      }
    }
  }
  
  return true;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::handlePubRequest(int fd , int type , void * data , int length)//实现了转发
{
  for (IPCEventChannels_t::iterator it = channels_m.begin(); it != channels_m.end();it ++)
  {
    IPCEventChannel & channel = *it;
    if (channel.getEventChannelId() == type)
    {
      IPCSubscribers_t & subscribers = channel.getSubscribers();
      for (IPCSubscribers_t::iterator pit = subscribers.begin(); pit != subscribers.end();pit ++)
      {
        int len = length + MESSAGE_HDR_LENGTH;
        int cmd = IPC_EVENT_TYPE_PUBLISH;
        int offset = 0;
        char buffer[MESSAGE_BD_MAX_LENGTH + MESSAGE_HDR_LENGTH] = {0};
        
        offset += IPCMsgUtils::compile(buffer , offset , len);
        offset += IPCMsgUtils::compile(buffer , offset , cmd);
        offset += IPCMsgUtils::compile(buffer , offset , type);
        offset += IPCMsgUtils::compile(buffer , offset , data , length);
        
        (void)deliverMsg(*pit , buffer , len);
      }
      
      return true;
    }
  }
  
  return true;
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::handleSubRequest(int fd , int type)
{
  return addChannel(type , fd);
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::handleUnsubRequest(int fd , int type)
{
  return updateChannelByType(type , fd);
}

/*
 *******************************************************************************
 *
 * @param Implementation Notes:
 *        None.
 *
 *******************************************************************************
 */
bool IPCPeerManager::deliverMsg(int fd , void * data , int length)
{
  int bytes = 0,ErrorV=0;
  do
  {
    int ret = send(fd , (char *)data + bytes , length - bytes , 0);
    if (ret == -1)
    {
    	ErrorV=errno;
      if (ErrorV == EINTR||ErrorV == EAGAIN)
      {
        continue;
      }
      else
      {
        
        LOGE(TAG, "socket send failed(%s)!!\n", strerror(errno));
        return false;
      }
    }
    
    bytes += ret;
  
  } while (bytes < length);
  
  return true;
}




